package mqtt

import akka.NotUsed
import akka.actor.typed._
import akka.actor.typed.receptionist.{Receptionist, ServiceKey}
import akka.actor.typed.scaladsl.{ActorContext, Behaviors}
import mqtt.EdgeDomain._
import mqtt.MqttUtils._
import org.slf4j.Logger

import java.sql.Timestamp
import java.util.Date
import scala.concurrent.ExecutionContextExecutor
import scala.concurrent.duration.DurationInt
import scala.language.postfixOps
import scala.util.Random

object DataAggregator {

  // Unique identifier at the AKKA level => used for identifying a specific service
  val serviceKey: ServiceKey[AggregatorCommand] = ServiceKey[AggregatorCommand]("dataAggregator")

  def apply(): Behavior[AggregatorCommand] = active(
    Map.empty[String, ActorRef[SensorCommand]],
    Map.empty[String, Report]
  )

  def active(sensors: Map[String, ActorRef[SensorCommand]],
             latestReports: Map[String, Report]): Behavior[AggregatorCommand] =
    Behaviors.receive[AggregatorCommand] {
      (context: ActorContext[AggregatorCommand], message: AggregatorCommand) =>
        val logger: Logger = context.log

        message match {
          case SensorRegister(deviceId, deviceDomain, deviceType, userId) =>
            logger.info(s"[${context.self.path.name}] Register new sensor: $message")
            // Register within aggregator
            val newSensor: ActorRef[SensorCommand] = context.spawn(
              EdgeDetector(deviceId, userId, deviceType),
              deviceDomain
            )
            val updatedSensors: Map[String, ActorRef[SensorCommand]] = sensors + (deviceId -> newSensor)
            active(updatedSensors, latestReports)
          case SensorReading(deviceId, sensorData) =>
            logger.info(s"[${context.self.path.name}] Received sensor heartbeat: $message")
            val timestamp = new Timestamp(new Date().getTime)
            val report: Report = sensorData match {
              case mq2Data: MQ2 =>
                MQ2Report(
                  properties = mq2Data,
                  event_time = getEventTimestamp(timestamp)
                )
              case dht11Data: DHT11 =>
                DHT11Report(
                  properties = dht11Data,
                  event_time = getEventTimestamp(timestamp)
                )
              case bmp280Data: BMP280 =>
                BMP280Report(
                  properties = bmp280Data,
                  event_time = getEventTimestamp(timestamp)
                )
              case _ => throw new RuntimeException("Un-supported type of sensor data.")
            }

            // Report updates
            val updatedReports: Map[String, Report] = latestReports + (deviceId -> report)
            logger.info(s"[${context.self.path.name}] Latest reports: " +
              s"${updatedReports.mkString("{\n\t", ",\n\t", "\n}")}")

            active(sensors, updatedReports)
          case SensorReporting(deviceId, replyTo) =>
            logger.info(s"[${context.self.path.name}] Sensor reporting task: $SensorReporting")
            val latestReport: Report = latestReports(deviceId)
            replyTo ! latestReport
            Behaviors.same
          case SensorUnregister(deviceId) =>
            logger.info(s"[${context.self.path.name}] Unregister sensor: $message")
            active(sensors - deviceId, latestReports - deviceId)
      }
    }
}

object EdgeDetector extends EdgeDomainJsonProtocol {

  def apply(deviceId: String, userId: String, deviceType: String): Behavior[SensorCommand] =
    Behaviors.setup { context: ActorContext[SensorCommand] =>
      // use a message adapter to turn a receptionist listing into a SensorCommand
      val receptionistSubscriber: ActorRef[Receptionist.Listing] = context.messageAdapter {
        case DataAggregator.serviceKey.Listing(set: Set[ActorRef[AggregatorCommand]]) =>
          ChangeAggregator(set.headOption)
      }

      // first subscribe to the receptionist using the service key
      context.system.receptionist ! Receptionist.Subscribe(DataAggregator.serviceKey, receptionistSubscriber)

      active(deviceId, userId, deviceType, None)
    }

  def sensorReading(id: String, user: String, deviceType: String): Sensor = {
    // Generate random data, simulate sensor reading
    deviceType match {
      case "mq2" =>
        val smokeReading: Double = Random.nextDouble() * 500
        val alarmReading: Boolean = if (smokeReading > 300) true else false

        MQ2(
          deviceId = id,
          userId = user,
          smokeConcentration = smokeReading,
          alarm = alarmReading,
          status = true
        )
      case "dht11" =>
        val tempReading: Double = Random.nextDouble() * 100
        val humidityReading: Double = Random.nextDouble() * 20

        DHT11(
          deviceId = id,
          userId = user,
          temperature = tempReading,
          humidity = humidityReading,
          status = true
        )
      case "bmp280" =>
        BMP280(
          deviceId = id,
          userId = user,
          temperature = 36,
          atmosphericPressure = 986,
          altitude = 150,
          status = true
        )
    }
  }

  def active(deviceId: String,
             userId: String,
             deviceType: String,
             aggregator: Option[ActorRef[SensorReading]]): Behavior[SensorCommand] =
    Behaviors.receive[SensorCommand] { (context: ActorContext[SensorCommand], message: SensorCommand) =>
      val logger: Logger = context.log

      message match {
        case SensorHeartBeat =>
          logger.info(s"[${context.self.path.name}] Reporting.")
          aggregator match {
            case Some(agg) =>
              val sensorValue: Sensor = sensorReading(deviceId, userId, deviceType)
              agg ! SensorReading(
                deviceId,
                sensorValue
              )
            case None => ()
          }
          Behaviors.same
        case ChangeAggregator(newAggregator) =>
          active(deviceId, userId, deviceType, newAggregator)
      }
    }
}

object MqttSimulator {

  def controller(aggregator: ActorRef[AggregatorCommand]): Behavior[NotUsed] =
    Behaviors.setup { context: ActorContext[NotUsed] =>
      implicit val ec: ExecutionContextExecutor = context.executionContext

      val deviceTypes: List[String] = List("mq2", "dht11", "bmp280")
      val devices: List[ActorRef[SensorCommand]] = deviceTypes.flatMap { deviceType: String =>
        val credentials: List[Credential] = readCredentials(deviceType)
        credentials.map { credential: Credential =>
          // register within aggregator
          aggregator ! SensorRegister(
            credential.device_id,
            credential.device_domain,
            deviceType,
            credential.user_id
          )

          context.spawn(EdgeDetector(credential.device_id, credential.user_id, deviceType), credential.device_domain)
        }
      }

    // send heartbeats every second
    context.system.scheduler.scheduleAtFixedRate(1 second, 5 second) { () =>
      devices.foreach((_: ActorRef[SensorCommand]) ! SensorHeartBeat)
    }

    Behaviors.empty[NotUsed]
  }

  val rootBehavior: Behavior[NotUsed] = Behaviors.setup { context: ActorContext[NotUsed] =>
    // initiate aggregator
    val dataAgg1: ActorRef[AggregatorCommand] = context.spawn(DataAggregator(), "dataAgg_1")
    context.system.receptionist ! Receptionist.register[AggregatorCommand](DataAggregator.serviceKey, dataAgg1)

    // controller for the sensors
    context.spawn(controller(dataAgg1), "controller")

    // change data aggregator after 10s
    Thread.sleep(10000)

    // unregister former aggregator and register a new one, we can update aggregator configuration along with it
    context.system.receptionist ! Receptionist.deregister[AggregatorCommand](DataAggregator.serviceKey, dataAgg1)
    val dataAgg2: ActorRef[AggregatorCommand] = context.spawn(DataAggregator(), "dataAgg_2")
    context.system.receptionist ! Receptionist.register[AggregatorCommand](DataAggregator.serviceKey, dataAgg2)

    Behaviors.empty[NotUsed]
  }

  def main(args: Array[String]): Unit = {
    implicit val system: ActorSystem[NotUsed] = ActorSystem(rootBehavior, "IoTSystem")
    implicit val ec: ExecutionContextExecutor = system.dispatchers.lookup(DispatcherSelector.default())

    system.scheduler.scheduleOnce(20 seconds, () => system.terminate())
  }
}
